package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;

public class Demo2Api {
    private Connection connection;

    @Before
    public void startup() {
        Configuration configuration = new Configuration();

        //指定zk的连接地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        //创建连接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void createTable() throws Exception {
        Admin admin = connection.getAdmin();


        //创建表的描述是对象，指定表名
        HTableDescriptor student = new HTableDescriptor(TableName.valueOf("student"));

        HColumnDescriptor info = new HColumnDescriptor("info");

        //增加列簇
        student.addFamily(info);

        //创建表
        admin.createTable(student);
    }

    @Test
    public void deleteTable() throws Exception {
        Admin admin = connection.getAdmin();
        //禁用表
        admin.disableTable(TableName.valueOf("student"));
        //删除表
        admin.deleteTable(TableName.valueOf("student"));
    }

    @Test
    public void put() throws IOException {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));

        //创建put对象，指定rowkey
        Put put = new Put("002".getBytes());

        //增加列
        put.addColumn("info".getBytes(), "name".getBytes(), "zhangsan".getBytes());

        //可以为列指定自定义版本号，注意版本号不能冲突，最好版本号递增

        //尽量使用基本数据类型代替字符串， 节省存储空间
        put.addColumn("info".getBytes(), "age".getBytes(), 1, Bytes.toBytes(23));

        student.put(put);
    }

    @Test
    public void get() throws Exception {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));

        Get get = new Get("002".getBytes());

        //查询数据返回result
        Result result = student.get(get);

        //解析result
        byte[] bname = result.getValue("info".getBytes(), "name".getBytes());
        String name = Bytes.toString(bname);

        //怎么存的数据就怎么取数据
        byte[] bage = result.getValue("info".getBytes(), "age".getBytes());
        int age = Bytes.toInt(bage);

        System.out.println(name + "\t" + age);
    }

    @Test
    public void cells() throws Exception {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));

        Get get = new Get("002".getBytes());

        //查询数据返回result
        Result result = student.get(get);

        //获取所有数据
        List<Cell> cells = result.listCells();

        //Cell 代表的是一个单元格，一个列的数据
        for (Cell cell : cells) {


            //列簇
            String family = Bytes.toString(CellUtil.cloneFamily(cell));


            //列名
            String quali = Bytes.toString(CellUtil.cloneQualifier(cell));


            //rowkey
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));


            if (quali.equals("name")) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(rowkey + "\t" + family + "\t" + quali + "\t" + value);
            } else if (quali.equals("age")) {
                int value = Bytes.toInt(CellUtil.cloneValue(cell));
                System.out.println(rowkey + "\t" + family + "\t" + quali + "\t" + value);
            }

        }
    }


    @Test
    public void putAll() throws Exception {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));


        //读取学生表数据
        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];


            //构建put 以学号作为rowkey
            Put put = new Put(id.getBytes());


            //增加列
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            //插入一条数据
            student.put(put);

        }

    }


    @Test
    public void scan() throws Exception {
        //获取表的示例对象
        Table student = connection.getTable(TableName.valueOf("student"));

        Scan scan = new Scan();

        //指定扫描的行数
//        scan.setLimit(10);

        //指定扫描的范围
        scan.withStartRow("1500100010".getBytes());
        scan.withStopRow("1500100020".getBytes());

        // 扫描表，默认是全表扫描
        ResultScanner scanner = student.getScanner(scan);

        //遍历返回的集合
        Result result;

        while ((result = scanner.next()) != null) {
            String id = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


    @After
    public void close() {
        try {

            if (connection != null)
                connection.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
